package cn.kgc.app

import cn.kgc.bean.Ads_log
import cn.kgc.utils.{JdbcUtil, MyKafkaUtil}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.text.SimpleDateFormat
import java.util.Date

/**
 * 需求:描述：实时统计每天各地区各城市各广告的点击总流量，并将其存入MySQL。
 *
 *   1）单个批次内对数据进行按照天维度的聚合统计;
 *   2）结合MySQL数据跟当前批次数据更新原有的数据。
 *
 *
 *
 */
object DateAreaCityAdCountHandler {
  def main(args: Array[String]): Unit = {

    // 创建时间格式化转换对象  把时间转化为年月日
    val dateformat = new SimpleDateFormat("yyyy-MM-dd")

    // 创建sparkStreaming 环境
    val conf = new SparkConf().setMaster("local[*]").setAppName("BlackListHandler")
    val ssc = new StreamingContext(conf,Seconds(3))
    // 1）读取Kafka数据之后，并对MySQL中存储的黑名单数据做校验；
    val source = MyKafkaUtil.getKafkaStream("ad", ssc)
    val adsLogDStream = source.map(x => {
      val arr = x.value().split(" ")

      Ads_log(arr(0).toLong, arr(1), arr(2), arr(3), arr(4))
    })

    // 1. 单个批次内对数据进行按照天维度的聚合统计;
    val reduceDStream = adsLogDStream.map(ads => {
      // 时间转化为年月日
      ((dateformat.format(new Date(ads.timestamp)), ads.area, ads.city, ads.adid), 1)
    }).reduceByKey(_+_)

   // 2 结合MySQL数据跟当前批次数据更新原有的数据。
    reduceDStream.foreachRDD(rdd =>{
      rdd.foreachPartition(it =>{
        // 获取连接对象
        val conn = JdbcUtil.getConnection
        it.foreach{
          case ((dt,area,city,adid),count) =>{
            JdbcUtil.executeUpdate(
              conn,
              """
                |insert into area_city_ad_count values(?,?,?,?,?)
                |ON DUPLICATE KEY
                |update count = count + ?
                |""".stripMargin,
              Array(dt,area,city,adid,count,count)
            )
          }
        }
        conn.close()
      })
    })


    ssc.start()
    ssc.awaitTermination()

  }


}
